package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionComment {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */
        //todo 3 从Kafka读取数据
        String topicName = "topic_db";
        String groupID = "user_register";
        tableEnv.executeSql("create table topic_db(\n" +
                " `database` string,\n" +
                " `table` string,\n" +
                " `type` string,\n" +
                " `data` map<string,string>,\n" +
                " `ts` bigint,\n" +
                " `pt` AS PROCTIME()\n" +
                ")"+ KafkaUtil.getKafkaDDL(topicName,groupID));

      //todo 4 读取课程评价表数据 review_info
        Table reviewInfo = tableEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['user_id'] user_id," +
                " data['course_id'] course_id," +
                " data['review_stars'] review_stars," +
                " if(data['review_stars']='5','五星好评','差评') review_level," +
                " date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                " data['create_time'] create_time," +
                " ts" +
                " from topic_db " +
                " where `table`='review_info' " +
                " and `type`='insert'");
        tableEnv.createTemporaryView("review_info",reviewInfo);
        //reviewInfo.execute().print();
        //todo 5 创建kafka dwd-interaction-comment表
        tableEnv.executeSql("create table dwd_interaction_comment(" +
                " id string," +
                " user_id string," +
                " course_id string," +
                " review_stars string," +
                " review_level string," +
                " data_id string," +
                " create_time string," +
                " ts bigint" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_interaction_comment"));



        //todo 6 将数据写入到kafka表
        tableEnv.executeSql("insert into dwd_interaction_comment select * from review_info");
    }
}
